Dataframes¶
- Dataframes are a special type of RDDs.
- Dataframes store two dimensional data, similar to the type of data stored in a spreadsheet.
- Each column in a dataframe can have a different type.
- Each row contains a
record.
- Similar to, but not the same as, pandas dataframes and R dataframes
import findspark
findspark.init()
from pyspark import SparkContext
sc = SparkContext(master="local[4]")
sc.version
# Just like using Spark requires having a SparkContext, using SQL requires an SQLContext
sqlContext = SQLContext(sc)
sqlContext
Constructing a DataFrame from an RDD of Rows¶
Each Row defines it's own fields, the schema is inferred.
# One way to create a DataFrame is to first define an RDD from a list of rows
some_rdd = sc.parallelize([Row(name=u"John", age=19),
Row(name=u"Smith", age=23),
Row(name=u"Sarah", age=18)])
some_rdd.collect()
# The DataFrame is created from the RDD or Rows
# Infer schema from the first row, create a DataFrame and print the schema
some_df = sqlContext.createDataFrame(some_rdd)
some_df.printSchema()
# A dataframe is an RDD of rows plus information on the schema.
# performing **collect()* on either the RDD or the DataFrame gives the same result.
print type(some_rdd),type(some_df)
print 'some_df =',some_df.collect()
print 'some_rdd=',some_rdd.collect()
Defining the Schema explicitly¶
The advantage of creating a DataFrame using a pre-defined schema allows the content of the RDD to be simple tuples, rather than rows.
# In this case we create the dataframe from an RDD of tuples (rather than Rows) and provide the schema explicitly
another_rdd = sc.parallelize([("John", 19), ("Smith", 23), ("Sarah", 18)])
# Schema with two fields - person_name and person_age
schema = StructType([StructField("person_name", StringType(), False),
StructField("person_age", IntegerType(), False)])
# Create a DataFrame by applying the schema to the RDD and print the schema
another_df = sqlContext.createDataFrame(another_rdd, schema)
another_df.printSchema()
# root
# |-- age: binteger (nullable = true)
# |-- name: string (nullable = true)
Loading DataFrames from disk¶
There are many maethods to load DataFrames from Disk. Here we will discuss three of these methods
- JSON
- CSV
- Parquet
In addition, there are API's for connecting Spark to an external database. We will not discuss this type of connection in this class.
Loading dataframes from JSON files¶
JSON is a very popular readable file format for storing structured data.
Among it's many uses are twitter, javascript communication packets, and many others. In fact this notebook file (with the extension .ipynb is in json format. JSON can also be used to store tabular data and can be easily loaded into a dataframe.
# when loading json files you can specify either a single file or a directory containing many json files.
path = "../../Data/people.json"
!cat $path
# Create a DataFrame from the file(s) pointed to by path
people = sqlContext.read.json(path)
print 'people is a',type(people)
# The inferred schema can be visualized using the printSchema() method.
people.show()
people.printSchema()
#load a Parquet file
print parquet_file
df = sqlContext.read.load(parquet_file)
df.show()
df2=df.select("name", "favorite_color")
df2.show()
outfilename="namesAndFavColors.parquet"
!rm -rf $dir/$outfilename
df2.write.save(dir+"/"+outfilename)
!ls -ld $dir/$outfilename
Loading a dataframe from a pickle file¶
Here we are loading a dataframe from a pickle file stored on S3. The pickle file contains meterological data that we will work on in future classes.
#List is a list of Rows. Stored as a pickle file.
df=sqlContext.createDataFrame(List)
print df.count()
df.show(1)
#selecting a subset of the rows so it fits in slide.
df.select('station','year','measurement').show(5)
### Save dataframe as Parquet repository
filename='%s/US_Weather_%s.parquet'%(data_dir,file_index)
!rm -rf $filename
df.write.save(filename)
- Parquet repositories are usually directories with many files.
- Parquet uses its column based format to compress the data.
!du -sh $filename
!du -sh $data_dir/$zip_file
Dataframe operations¶
Spark DataFrames allow operations similar to pandas Dataframes. We demonstrate some of those. For more, see this article
df.describe().select('station','elevation','measurement').show()
df.groupby('measurement').agg({'year': 'min', 'station':'count'}).show()
Using SQL queries on DataFrames¶
There are two main ways to manipulate DataFrames:
Imperative manipulation¶
Using python methods such as .select and .groupby.
- Advantage: order of operations is specified.
- Disrdavantage : You need to describe both what is the result you want and how to get it.
Declarative Manipulation (SQL)¶
- Advantage: You need to describe only what is the result you want.
- Disadvantage: SQL does not have primitives for common analysis operations such as covariance
people.show()
# Register this DataFrame as a table.
people.registerTempTable("people")
# SQL statements can be run by using the sql methods provided by sqlContext
teenagers = sqlContext.sql("SELECT name FROM people WHERE age >= 13 AND age <= 19")
for each in teenagers.collect():
print(each[0])
Counting the number of occurances of each measurement, imparatively¶
L=df.groupBy('measurement').count().collect()
D=[(e.measurement,e['count']) for e in L]
sorted(D,key=lambda x:x[1], reverse=False)[:6]
Counting the number of occurances of each measurement, declaratively.¶
sqlContext.registerDataFrameAsTable(df,'weather') #using older sqlContext instead of newer (V2.0) sparkSession
query='SELECT measurement,COUNT(measurement) AS count FROM weather GROUP BY measurement ORDER BY count'
print query
sqlContext.sql(query).show()
Performing a map command¶
In order to perform map, you need to first transform the dataframe into an RDD.
df.rdd.map(lambda row:(row.longitude,row.latitude)).take(5)
Approximately counting the number of distinct elements in column¶
import pyspark.sql.functions as F
F.approx_count_distinct?
df.agg({'station':'approx_count_distinct','year':'min'}).show()
Approximate Quantile¶
The method .approxQuantile computes the approximate quantiles.
Recall that this is how we computed the pivots for the distributed bucket sort.
print 'with accuracy 0.1: ',df.approxQuantile('year', [0.1*i for i in range(1,10)], 0.1)
print 'with accuracy 0.01: ',df.approxQuantile('year', [0.1*i for i in range(1,10)], 0.01)
Lets collect the exact number of rows for each year¶
This will take much longer than ApproxQuantile on a large file
# Lets collect the exact number of rows for each year ()
query='SELECT year,COUNT(year) AS count FROM weather GROUP BY year ORDER BY year'
print query
counts=sqlContext.sql(query)
A=counts.toPandas()
A.head()
import pandas as pd
A.plot.line('year','count')
grid()
Reading rows selectively¶
Suppose we are only interested in snow measurements. We can apply an SQL query directly to the parquet files. As the data is organized in columnar structure, we can do the selection efficiently without loading the whole file to memory.
Here the file is small, but in real applications it can consist of hundreds of millions of records. In such cases loading the data first to memory and then filtering it is very wasteful.
query='SELECT station,measurement,year FROM weather WHERE measurement="SNOW"'
print query
df2 = sqlContext.sql(query)
print df2.count(),df2.columns
df2.show(5)